package com.chencong.online.function;

import com.chencong.online.bean.OrderEventBean;
import com.chencong.online.bean.OrderResultBean;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import sun.security.x509.CertificateX509Key;

/**
 * @Author chencong
 * @Description自定义函数监控订单超时
 * @Date 9:37 下午 2022/1/5
 * @Param
 **/
public class OrderPayDetectProcessFunc extends KeyedProcessFunction<Long, OrderEventBean, OrderResultBean> {
    // 定义状态，保存之前点单是否已经来过create、pay的事件
    ValueState<Boolean> isPayedState;
    ValueState<Boolean> isCreatedState;

    // 定义状态，保存定时器时间戳
    ValueState<Long> timerTsState;

    //侧输出流
    OutputTag<OrderResultBean> outputTag = new OutputTag<OrderResultBean>("order-timeout") {
    };

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化状态
        isPayedState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("is-payed", Boolean.class, false));

        isCreatedState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("is-created", Boolean.class, false));

        timerTsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-ts", long.class));

    }

    @Override
    public void processElement(OrderEventBean value, Context ctx, Collector<OrderResultBean> out) throws Exception {
        //获取当前状态
        Boolean isPayed = isPayedState.value();
        Boolean isCreated = isCreatedState.value();
        Long timerTs = timerTsState.value();
        //判断当前事件类型
        if ("create".equals(value.getEventType())) {
            //1、如果来的是create，要判断是否支付过
            if (isPayed) {
                //1.1、如果已经正常支付，输出正常匹配结果
                out.collect(new OrderResultBean(value.getOrderId(), "支付成功！！！"));
                //清空状态，删除定时器
                isPayedState.clear();
                isCreatedState.clear();
                timerTsState.clear();
                ctx.timerService().deleteEventTimeTimer(timerTs);
            } else {
                //1.2、如果没有支付过、注册一个15分钟的定时器，开始等待支付事件
                long ts = (value.getTimestamp() + 15 * 60) * 1000L;
                ctx.timerService().registerEventTimeTimer(ts);
                //更新触发时间
                timerTsState.update(ts);
                //更新支付状态
                isCreatedState.update(true);
            }

        } else if ("pay".equals(value.getEventType())) {
            //2、如果来的是pay，要判断是否有下单事件来过
            if (isCreated) {
                //2.1、已经有过下单事件，要继续判断支付的时间戳是否超过15分钟
                if (value.getTimestamp() * 1000L < timerTs) {
                    //2.1.1、当在15分钟以内时，没有超时，正常输出
                    out.collect(new OrderResultBean(value.getOrderId(), "支付成功！！！"));
                } else {
                    //2.1.2、已经超时，输出侧输出流报警
                    ctx.output(outputTag, new OrderResultBean(value.getOrderId(), "订单超时"));
                }
                //统一清空状态
                isCreatedState.clear();
                isPayedState.clear();
                timerTsState.clear();
                ctx.timerService().deleteEventTimeTimer(timerTs);
            } else {
                //2.2、没有下单事件，加入有乱序，注册一个定时器等待下单事件

                ctx.timerService().registerEventTimeTimer(value.getTimestamp() * 1000L);

                //更新状态
                timerTsState.update(value.getTimestamp() * 1000L);
                isPayedState.update(true);

            }
        }

    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<OrderResultBean> out) throws Exception {
        //当定时器触发时，说明一定有一个事件没来
        if (isPayedState.value()) {
            //此刻pay来了，create没有来
            ctx.output(outputTag, new OrderResultBean(ctx.getCurrentKey(), "pay来了，但是没有create日志"));
        } else {
            //此时pay没来，支付超时
            ctx.output(outputTag, new OrderResultBean(ctx.getCurrentKey(), "超时"));
        }
        //清空状态
        isPayedState.clear();
        isCreatedState.clear();
        timerTsState.clear();

    }
}
